package cn.admin.remoting.impl;

import cn.admin.remoting.*;
import org.apache.mina.common.IoConnector;
import org.apache.mina.common.IoFutureListener;
import org.apache.mina.common.IoSession;
import org.apache.mina.common.WriteFuture;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

public class DefaultConnection implements Connection {

    private static final String SESSION_CONN = "session2connection";
    private final IoSession session;
    private final ConnectionFactory factory;
    private final boolean isClientSide;
    private byte lastRequestProtocol = -1;
    private final CopyOnWriteArrayList<IOEventListener> ioListeners = new CopyOnWriteArrayList<>();
    private final IOEventListener.ConnectionMsgReceivedListener msgListener;

    public DefaultConnection(IoSession session,ConnectionFactory factory) {
        if (session == null) {
            throw new IllegalArgumentException("session is null");
        } else {
            this.session = session;
            this.factory = factory;
            this.session.setAttribute(SESSION_CONN,this);
            this.session.setWriteTimeout(10);
            this.isClientSide = this.session.getService() instanceof IoConnector;
            this.msgListener = new DefaultMsgListener();
            this.ioListeners.add(this.msgListener);
        }
    }

    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
        in.defaultReadObject();
        this.session.setAttribute(SESSION_CONN,this);
    }

    public static DefaultConnection findBySession(IoSession session) {
        return (DefaultConnection) session.getAttribute(SESSION_CONN);
    }

    public IoSession getIoSession() {
        return this.session;
    }

    @Override
    public boolean isConnected() {
        return this.session != null && !this.session.isClosing();
    }

    @Override
    public int getRemotePort() {
        return ((InetSocketAddress)this.session.getRemoteAddress()).getPort();
    }

    @Override
    public String getRemoteAddress() {
        InetSocketAddress remoteSocket = (InetSocketAddress) this.session.getRemoteAddress();
        return remoteSocket.getAddress().getHostAddress();
    }

    @Override
    public String getRemoteSocket() {
        return this.getRemoteAddress() + ":" + this.getRemotePort();
    }

    @Override
    public int getLocalPort() {
        return ((InetSocketAddress)this.session.getLocalAddress()).getPort();
    }

    @Override
    public String getLocalAddress() {
        InetSocketAddress localSocket = (InetSocketAddress) this.session.getLocalAddress();
        return localSocket.getAddress().getHostAddress();
    }

    @Override
    public String getLocalSocket() {
        return this.getLocalAddress() + ":" + this.getLocalPort();
    }

    @Override
    public boolean isClientSide() {
        return this.isClientSide;
    }

    @Override
    public Client getClient() {
        return DefaultClient.findBySession(session);
    }

    @Override
    public ConnectionFactory getConnectionFactory() {
        return this.factory;
    }

    @Override
    public byte lastRequestProtocol() {
        return this.lastRequestProtocol;
    }

    @Override
    @Deprecated
    public void close() {
        if (this.session.isConnected()) {
            this.session.close();
        }
    }

    @Override
    public void close(String cause) {
        if (this.session.isConnected()) {
            this.session.setAttribute(CLOSE_REASON,cause);
            this.session.close();
        }
    }

    @Override
    public void write(Object connectionMsg) {
        this.write(connectionMsg,null);
    }

    @Override
    public WriteFuture write(Object connectionMsg, IoFutureListener wfl) {
        if (connectionMsg instanceof ConnectionRequest) {
            ((ConnectionRequest)connectionMsg).setClient(getClient());
        }

        if (OverFlowWriteFuture.getThreshold() > 0L) {
            long notFlushed = 0L;
            DefaultConnection conn;

            for (Connection connection : LCL_AliveSessionsListener.aliveSessions()) {
                conn = (DefaultConnection) connection;
                notFlushed += (long)conn.session.getScheduledWriteBytes();
            }

            if (notFlushed > OverFlowWriteFuture.getThreshold()) {
                return new OverFlowWriteFuture(this.session,wfl,notFlushed);
            }
        }
        WriteFuture wf = this.session.write(connectionMsg);
        if (wfl != null) {
            wf.addListener(wfl);
        }
        return wf;
    }

    @Override
    public void addConnectionEvtListener(IOEventListener listener) {
        this.ioListeners.addIfAbsent(listener);
    }

    @Override
    public void removeConnectionEvtListener(IOEventListener listener) {
        this.ioListeners.remove(listener);
    }

    @Override
    public List<IOEventListener> getConnectionEvtListeners() {
        return this.ioListeners;
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> T setAttribute(String key, T value) {
        return (T) this.session.setAttribute(key,value);
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> T getAttribute(String key) {
        return (T) this.session.getAttribute(key);
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T> T removeAttribute(String key) {
        return (T) this.session.removeAttribute(key);
    }

    public IOEventListener.ConnectionMsgReceivedListener getMsgReceivedListener() {
        return this.msgListener;
    }

    public void setLastRequestProtocol(byte lastRequestProtocol) {
        this.lastRequestProtocol = lastRequestProtocol;
    }

    @Override
    public String toString() {
        return ("L=" + this.getLocalSocket()) +
                ", R=" + this.getRemoteSocket() +
                ", S=" + (this.isConnected() ? "Connected" : "DisConnected");
    }
}
